package com.whale.springboot.kafka.controller;

import com.whale.springboot.kafka.service.ProducerService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@RestController
@RequestMapping("/producer")
@Slf4j
public class KafkaProduerController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private ProducerService kafkaProducerService;

    @RequestMapping("/sync")
    public String sendSync(@RequestParam String msg) throws ExecutionException, InterruptedException, TimeoutException {
        // 同步发送，如果发送到broker卡着的话，这里也会阻塞，可以设置等待10s，10s之后就直接报错
        kafkaTemplate.send("whale", msg).get(10, TimeUnit.SECONDS);

        // 这里有异常的话，消息还是发出去了，所以可以采用事务
        return "OK";
    }

    @RequestMapping("/testConsumerException")
    public String testConsumerException3() throws ExecutionException, InterruptedException, TimeoutException {
        // 同步发送，如果发送到broker卡着的话，这里也会阻塞，可以设置等待10s，10s之后就直接报错
        kafkaTemplate.send("hello-kafka-test-topic-exception", "1").get(10, TimeUnit.SECONDS);
        return "OK";
    }

    @RequestMapping("/testConsumerBatch")
    public String testConsumerBatch() throws ExecutionException, InterruptedException, TimeoutException {
        //写入多条数据到批量topic：hello-batch
        String topic = "hello-batch";
        for(int i = 0; i < 20; i++) {
            kafkaProducerService.sendMessageSync(topic, null, "batchMessage" + i);
        }
        return "OK";
    }

    @RequestMapping("/testSendMessageInTransaction")
    public String testSendMessageInTransaction() throws ExecutionException, InterruptedException, TimeoutException {
        //写入多条数据到批量topic：hello-batch
        String topic = "hello-kafka-test-topic";
        String key = "key1";
        String message = "helloSendMessageInTransaction";
        for(int i = 0; i < 20; i++) {
            kafkaProducerService.sendMessageInTransaction(topic, key, message);
        }
        return "OK";
    }

    @RequestMapping("/testConsumerInterceptor")
    public String testConsumerInterceptor() throws ExecutionException, InterruptedException, TimeoutException {
        String topic = "consumer-interceptor";
        for(int i = 0; i < 2; i++) {
            kafkaProducerService.sendMessageSync(topic,null, "normalMessage" + i);
        }
        kafkaProducerService.sendMessageSync(topic, null, "filteredMessage");
        kafkaProducerService.sendMessageSync(topic, null, "filterMessage");
        return "OK";
    }

    @RequestMapping("/testFenqu")
    public String testFenqu() throws ExecutionException, InterruptedException, TimeoutException {
        String topic = "jkhtest6";
        for (int i = 0; i < 100; i++) {
            kafkaProducerService.sendMessageSync(topic, null, "我jkhtest" + i);
            log.info("发送了~" + i);
        }
        return "OK";
    }

    @RequestMapping("/testFenqu2")
    @Transactional
    public String testFenqu2() throws ExecutionException, InterruptedException, TimeoutException {
        String topic = "jkhtest6";
        for (int i = 0; i < 100; i++) {
            kafkaProducerService.sendMessageSync(topic, null, "我jkhtest" + i);
            log.info("发送了~" + i);
        }
        return "OK";
    }
}